import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Callable, Any, List, Tuple

class GoRoutines:
    def __init__(self, max_threads: int = 10):
        """初始化线程池"""
        self.executor = ThreadPoolExecutor(max_threads)

    async def run_async(self, coro: Callable[..., Any], *args, **kwargs) -> Any:
        """运行异步任务"""
        return await coro(*args, **kwargs)

    async def run_blocking(self, func: Callable[..., Any], *args, **kwargs) -> Any:
        """运行阻塞任务"""
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(self.executor, func, *args, **kwargs)

    async def run_tasks(self, tasks: List[asyncio.Task]) -> List[Any]:
        """并发运行多个任务"""
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

    def close(self):
        """关闭线程池"""
        self.executor.shutdown(wait=True)

# 全局任务管理器
goroutines = GoRoutines()

# 单任务接口
async def go(func: Callable[..., Any], *args, **kwargs) -> Any:
    """运行单个异步或阻塞任务"""
    if asyncio.iscoroutinefunction(func):
        return await goroutines.run_async(func, *args, **kwargs)
    else:
        return await goroutines.run_blocking(func, *args, **kwargs)

# 并发任务接口
async def go_all(tasks: List[Tuple[Callable[..., Any], List[Any], dict]]) -> List[Any]:
    """并发运行多个任务"""
    async_tasks = [
        go(task, *args, **kwargs) for task, args, kwargs in tasks
    ]
    return await goroutines.run_tasks(async_tasks)

def shutdown():
    """关闭全局资源"""
    goroutines.close()



# ---------------------------- 以下为说明 ----------------------------
'''
1. tasks 格式固定为三元组：
    每个任务必须是 (function, args_list, kwargs_dict) 格式。
    这是为了区分协程函数和普通函数，并正确传递参数。

    # 定义一个测试函数
    def test_function(a, b, c=0, d=0):
        return a + b + c + d
    
    # 定义任务
    task = (
        test_function,        # 要执行的函数
        [1, 2],               # 位置参数
        {'c': 3, 'd': 4}      # 关键字参数
    )


2. go 方法检查任务类型：
    如果是协程函数，则用 await coro(...) 执行；
    如果是阻塞函数，则用 loop.run_in_executor(...) 包装执行。
3. RuntimeWarning 修复：
    原因是协程未被正确 await。修复后，确保所有协程通过 await 调用。
4. 捕获异常：
    go_all 的 asyncio.gather 使用 return_exceptions=True，避免异常任务阻塞整个程序。
'''


# ---------------------------- 以下为测试代码 ----------------------------


import asyncio
from time import sleep

# 异步任务
async def async_task(name: str, delay: int):
    print(f"--> Async Task {name} started")
    print(f"    --> Task {name} will sleep for {delay} seconds")
    await asyncio.sleep(delay)
    print(f"<-- Async Task {name} finished")
    return f"<-->: Result from {name}"

# 阻塞任务
def blocking_task(name: str, delay: int):
    print(f"--> Blocking Task {name} started")
    print(f"    --> Task {name} will sleep for {delay} seconds")
    sleep(delay)
    print(f"<-- Blocking Task {name} finished")
    return f"<-->: Result from {name}"

# 并发运行多个任务
async def main():
    tasks = [
        (async_task, ["Async-1", 2], {}),  # 异步任务
        (blocking_task, ["Blocking-1", 3], {}),  # 阻塞任务
        (async_task, ["Async-2", 1], {}),  # 异步任务
        (blocking_task, ["Blocking-2", 2], {}),  # 阻塞任务
    ]
    results = await go_all(tasks)
    print("All Results:", results)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    finally:
        shutdown()
